package cn.kennylee.codehub.mq.lip.integration.rocketmq.producer;

import cn.hutool.core.util.StrUtil;
import cn.hutool.extra.spring.SpringUtil;
import cn.hutool.json.JSONUtil;
import cn.kennylee.codehub.mq.lip.integration.base.AbstractMqChannel;
import cn.kennylee.codehub.mq.lip.integration.base.MessageVo;
import cn.kennylee.codehub.mq.lip.integration.rocketmq.RocketMqConfigProperties;
import cn.kennylee.codehub.mq.lip.integration.rocketmq.RocketMqListenerRegistrar;
import jakarta.annotation.Resource;
import lombok.Getter;
import lombok.Setter;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.producer.SendCallback;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.MQVersion;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.springframework.lang.NonNull;
import org.springframework.lang.Nullable;
import org.springframework.messaging.Message;

import java.time.Duration;
import java.util.List;
import java.util.Optional;
import java.util.concurrent.CompletableFuture;

/**
 * <p> RocketMQ消息生产者组件 </p>
 * <p>Created on 2025/4/15.</p>
 *
 * @author kennylee
 * @since 0.0.1
 */
@Setter
@Slf4j
public class RocketMqChannel extends AbstractMqChannel<SendResult> {

    /**
     * 时间单位常量（秒为基准）
     */
    private static final int SECONDS_PER_MINUTE = 60;
    private static final int SECONDS_PER_HOUR = SECONDS_PER_MINUTE * 60;
    private static final int MINUTES_PER_HOUR = 60;

    private RocketMQTemplate rocketMqTemplate;

    @Resource
    private RocketMqConfigProperties rocketMqConfigProperties;

    /**
     * 是否添加路由键后缀
     * <p>默认为true，表示在路由键后添加后缀</p>
     */
    @Getter
    private boolean addRouterKeySuffix = true;

    public RocketMqChannel(String channel, RocketMQTemplate rocketMqTemplate) {
        super(channel);
        this.rocketMqTemplate = rocketMqTemplate;
    }

    @Override
    public SendResult sendMsg(@NonNull String channel, @NonNull String routeKey, @NonNull MessageVo messageVo) {
        if (isSkipSend()) {
            log.info("测试环境，不发送消息：channel={}, routeKey={}, messageVo={}", channel, routeKey, convertToMsgJsonStr(messageVo));
            return new SendResult();
        }

        if (StrUtil.isBlank(messageVo.getHashKey())) {
            SendResult sendResult = rocketMqTemplate.syncSend(buildDestination(channel, routeKey), buildMessage(messageVo));
            log.info("发送消息[{}]结果，消息返回内容: {}", routeKey, JSONUtil.toJsonStr(sendResult));
            return sendResult;
        }

        SendResult sendResult = rocketMqTemplate.syncSendOrderly(buildDestination(channel, routeKey), buildMessage(messageVo), messageVo.getHashKey());
        log.info("发送消息[{}]结果，消息返回内容: {}", routeKey, JSONUtil.toJsonStr(sendResult));
        return sendResult;
    }

    @Override
    public SendResult sendMsg(@NonNull String channel, @NonNull String routeKey, @NonNull MessageVo messageVo, @NonNull Duration delayTime) {
        log.info("发送延迟消息：channel={},routeKey={}, messageVo={}, delayTime={}", getChannel(), routeKey, convertToMsgJsonStr(messageVo), delayTime);

        if (isSkipSend()) {
            log.info("测试环境，不发送消息：routeKey={}, messageVo={}, delayTime={}", routeKey, convertToMsgJsonStr(messageVo), delayTime);
            return new SendResult();
        }

        if (StrUtil.isBlank(messageVo.getHashKey())) {
            SendResult sendResult = doSyncSendDelayTime(buildDestination(channel, routeKey), buildMessage(messageVo), delayTime);
            log.info("发送延迟消息[{}]结果，消息返回内容: {}", routeKey, JSONUtil.toJsonStr(sendResult));
            return sendResult;
        }

        throw new UnsupportedOperationException("不支持顺序消息的延迟消息发送");
    }

    /**
     * 同步发送延迟消息（兼容 4.x 和 5.x）
     *
     * @param destination 消息目的地
     * @param message     消息内容
     * @param delayTime   延迟时间
     * @return 发送结果
     */
    protected SendResult doSyncSendDelayTime(String destination, Message<String> message, Duration delayTime) {
        // 5.0.0 版本以上，支持 setDelayTimeMs
        if (isDelayTimeMsSupported()) {
            return rocketMqTemplate.syncSendDelayTimeMills(destination, message, delayTime.toMillis());
        }
        log.debug("RocketMQ当前版本判断为不支持 setDelayTimeMs，使用延迟级别发送消息");
        // 4.x 用法：转换为延迟级别
        int delayLevel = toDelayLevel(delayTime);
        return rocketMqTemplate.syncSend(destination, message, rocketMqTemplate.getProducer().getSendMsgTimeout(), delayLevel);
    }

    /**
     * 将 Duration 转换为 RocketMQ 延迟级别
     *
     * @param delayTime 请求的延迟时间（必须非空）
     * @return 对应的 RocketMQ 延迟级别 (1~18)
     */
    private int toDelayLevel(@NonNull Duration delayTime) {
        List<Duration> levels = rocketMqConfigProperties.getDelay().getLevels();
        long requestMillis = delayTime.toMillis();
        String formattedRequest = formatDuration(delayTime);

        // 检查是否小于最小级别
        if (requestMillis <= levels.get(0).toMillis()) {
            log.info("目标延迟时间 {} ≤ 最小级别 {} (级别1)，直接使用级别1",
                formattedRequest, formatDuration(levels.get(0)));
            return 1;
        }

        // 精确匹配检查
        for (int i = 0; i < levels.size(); i++) {
            Duration currentLevel = levels.get(i);
            long levelMillis = currentLevel.toMillis();

            if (requestMillis == levelMillis) {
                if (log.isDebugEnabled()) {
                    log.debug("目标延迟时间 {} 精确匹配级别 {} [{}]",
                        formattedRequest, i + 1, formatDuration(currentLevel));
                }
                return i + 1;
            }

            if (requestMillis < levelMillis) {
                Duration prevLevel = i > 0 ? levels.get(i - 1) : null;

                // 计算与前后级别的差距
                long diffPrev = prevLevel != null ?
                    requestMillis - prevLevel.toMillis() : Long.MAX_VALUE;
                long diffNext = levelMillis - requestMillis;

                // 选择最接近的级别
                int chosenLevel = (diffPrev <= diffNext) ? i : i + 1;
                Duration closestLevel = (diffPrev <= diffNext) ? prevLevel : currentLevel;

                log.warn("目标延迟时间 {} 介于: {} (级别{}) 和 {} (级别{}) 之间，最终选择级别 {} [{}]",
                    formattedRequest, prevLevel != null ? formatDuration(prevLevel) : "N/A", i,
                    formatDuration(currentLevel), i + 1,
                    chosenLevel, formatDuration(closestLevel));

                return chosenLevel;
            }
        }

        // 超出最大级别的情况
        Duration maxLevel = levels.get(levels.size() - 1);
        log.warn("延迟时间 {} > 最大级别 {} (级别{})，强制使用最大级别",
            formattedRequest, formatDuration(maxLevel), levels.size());
        return levels.size();
    }

    /**
     * 格式化Duration为中文时间字符串
     *
     * @param duration 时间间隔（可为null）
     * @return 格式化后的字符串，如"5秒", "2分30秒", "1小时15分"
     */
    public static String formatDuration(Duration duration) {
        if (duration == null) {
            return "N/A";
        }

        final long totalSeconds = duration.getSeconds();

        // 秒级处理
        if (totalSeconds < SECONDS_PER_MINUTE) {
            return totalSeconds + "秒";
        }

        // 分钟级处理
        if (totalSeconds < SECONDS_PER_HOUR) {
            final long minutes = totalSeconds / SECONDS_PER_MINUTE;
            final long remainingSeconds = totalSeconds % SECONDS_PER_MINUTE;
            return minutes + "分" + (remainingSeconds > 0 ? remainingSeconds + "秒" : "");
        }

        // 小时级处理
        final long hours = totalSeconds / SECONDS_PER_HOUR;
        final long remainingMinutes = (totalSeconds % SECONDS_PER_HOUR) / SECONDS_PER_MINUTE;
        return hours + "小时" + (remainingMinutes > 0 ? remainingMinutes + "分" : "");
    }

    /**
     * 检查当前版本是否支持 setDelayTimeMs（5.0.0+）
     */
    public static boolean isDelayTimeMsSupported() {
        return MQVersion.CURRENT_VERSION >= MQVersion.Version.V5_0_0.ordinal();
    }

    @Override
    public CompletableFuture<SendResult> asyncSendMsg(@NonNull String channel, @NonNull String routeKey, @NonNull MessageVo messageVo) {
        log.info("异步发送消息：channel={}, routeKey={}, messageVo={}", getChannel(), routeKey, convertToMsgJsonStr(messageVo));

        if (isSkipSend()) {
            log.info("测试环境，不发送消息：routeKey={}, messageVo={}", routeKey, convertToMsgJsonStr(messageVo));
            return CompletableFuture.completedFuture(new SendResult());
        }

        final CompletableFuture<SendResult> future = new CompletableFuture<>();

        if (StrUtil.isBlank(messageVo.getHashKey())) {
            rocketMqTemplate.asyncSend(buildDestination(channel, routeKey), buildMessage(messageVo), new RocketMqSendCallback(future));
        } else {
            rocketMqTemplate.asyncSendOrderly(buildDestination(channel, routeKey), buildMessage(messageVo), messageVo.getHashKey(), new RocketMqSendCallback(future));
        }
        return future;
    }

    @Override
    public CompletableFuture<SendResult> asyncSendMsg(@NonNull String channel, @NonNull String routeKey, @NonNull MessageVo messageVo, @NonNull Duration delayTime) {
        throw new UnsupportedOperationException("不支持异步发送延迟消息");
    }

    /**
     * 构建消息最终mq地址
     *
     * @param routerKey 消息标签
     * @return 最终的mq地址
     */
    @NonNull
    private String buildDestination(@NonNull String routerKey) {
        return buildDestination(null, routerKey);
    }

    /**
     * 构建消息最终mq地址
     *
     * @param routerKey 消息标签
     * @return 最终的mq地址
     */
    @NonNull
    private String buildDestination(@Nullable String channel, @NonNull String routerKey) {
        String tag = addRouterKeySuffix ?
            // 添加路由键后缀
            RocketMqListenerRegistrar.addRouterKeySuffix(routerKey, SpringUtil.getApplicationContext()) :
            // 不添加路由键后缀
            routerKey;
        return Optional.ofNullable(channel).orElse(getChannel()) + ":" + tag;
    }

    /**
     * 异步发送消息回调
     *
     * @param future 异步结果
     */
    public record RocketMqSendCallback(CompletableFuture<SendResult> future) implements SendCallback {

        @Override
        public void onSuccess(SendResult sendResult) {
            if (log.isInfoEnabled()) {
                log.info("异步发送消息[msgId={}]结果，消息返回内容: {}", sendResult.getMsgId(), sendResult.toString());
            }
            future.complete(sendResult);
        }

        @Override
        public void onException(Throwable e) {
            log.error("异步发送消息失败：e={}", e.getMessage(), e);
            future.completeExceptionally(e);
        }
    }
}
